package com.atguigu.flink.exec2;

import com.atguigu.flink.pojo.LoginEvent;
import com.atguigu.flink.util.MyUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

import java.util.List;

/**
 * Created by Smexy on 2022/11/25
 *  如果同一用户（可以是不同IP）在2秒之内连续两次登录，都失败，就认为存在恶意登录的风险
 *  
 *  
 *  5402,83.149.11.115,success,1558430815
 *          用户，ip，登录标记，时间
 *
 *    思路:  开一个基于个数的窗口
 *              窗口只放两条数据，判断是否都是失败，以及时间间隔是否在2秒之内
 *
 *         登录的数据，必须是有序的！
 */
public class Demo5_Login
{
    public static void main(String[] args) {
        
        
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);
        
        env.readTextFile("data/LoginLog.csv")
           .map(new MapFunction<String, LoginEvent>()
           {
               @Override
               public LoginEvent map(String value) throws Exception {
                   String[] words = value.split(",");
                    return new LoginEvent(
                        Long.valueOf(words[0]),
                        words[1],
                        words[2],
                        Long.valueOf(words[3])
                    )  ;
               }
           })
            .keyBy(LoginEvent::getUserId)
            .countWindow(2,1)
            .process(new ProcessWindowFunction<LoginEvent, String, Long, GlobalWindow>()
            {
                @Override
                public void process(Long key, Context context, Iterable<LoginEvent> elements, Collector<String> out) throws Exception {

                    List<LoginEvent> eventList = MyUtil.parseList(elements);

                    if (eventList.size() == 2){
                        LoginEvent e1 = eventList.get(0);
                        LoginEvent e2 = eventList.get(1);

                        if ("fail".equals(e1.getEventType())
                            &&
                            "fail".equals(e2.getEventType())
                            &&
                            Math.abs(e1.getEventTime() - e2.getEventTime()) <= 2 ){

                            out.collect(key+"恶意登录....");

                        }
                    }
                }
            })
           .print().setParallelism(1);
        
        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }
        
    }
}
